package com.bw.test2;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class BaseLogApp {

    public static void main(String[] args) throws Exception {
        //1.创建流式环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.设置并行度
        env.setParallelism(1);
        //3.获取kafka  topic_log数据
        DataStreamSource<String> logDS =
                env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("topic_log", "topic_log_base_app"));
        //4.进行过滤
        SingleOutputStreamOperator<JSONObject> filterDs = logDS.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
                try {
                    //正常数据
                    JSONObject jsonObject = JSON.parseObject(value);
                    System.out.println(jsonObject);
                    out.collect(jsonObject);
                } catch (Exception e) {
                    //异常数据
                    System.out.println("异常数据：" + value + e.getMessage());
                }
            }
        });
        OutputTag<String> errTag = new OutputTag<String>("err") {
        };
        OutputTag<String> startTag = new OutputTag<String>("start") {
        };
        OutputTag<String> displayTag = new OutputTag<String>("display") {
        };
        OutputTag<String> actionTag = new OutputTag<String>("action") {
        };
        //5.进行分流
        SingleOutputStreamOperator<String> flDs = filterDs.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value, Context ctx, Collector<String> out) throws Exception {
                //1.错误数据  原封不动输出
                JSONObject err = value.getJSONObject("err");
                if (err != null) {
                    ctx.output(errTag, err.toJSONString());
                }
                value.remove("err");
                //2.err 这个字段删除掉
                //3.过滤出start数据  原封不动输出
                JSONObject start = value.getJSONObject("start");
                if (start != null) {
                    ctx.output(startTag, start.toJSONString());
                } else {
                    JSONObject common = value.getJSONObject("common");
                    JSONObject page = value.getJSONObject("page");
                    Long ts = value.getLong("ts");
                    //4.过滤出display
                    JSONArray displays = value.getJSONArray("displays");
                    for (int i = 0; i < displays.size(); i++) {
                        JSONObject jsonObject = displays.getJSONObject(i);
                        JSONObject jsonObject1 = new JSONObject();
                        jsonObject1.put("common", common);
                        jsonObject1.put("page", page);
                        jsonObject1.put("display", jsonObject);
                        jsonObject1.put("ts", ts);
                        ctx.output(displayTag, jsonObject1.toJSONString());
                    }
                    //5.过滤出action
                    JSONArray actions = value.getJSONArray("actions");
                    for (int i = 0; i < actions.size(); i++) {
                        JSONObject jsonObject = actions.getJSONObject(i);
                        JSONObject jsonObject1 = new JSONObject();
                        jsonObject1.put("common", common);
                        jsonObject1.put("page", page);
                        jsonObject1.put("action", jsonObject);
                        jsonObject1.put("ts", ts);
                        ctx.output(actionTag, jsonObject1.toJSONString());
                    }
                    value.remove("displays");
                    value.remove("actions");
                    out.collect(value.toJSONString());
                }
            }
        });



        //6.进行存入到对应的kafka中
        DataStream<String> errDs = flDs.getSideOutput(errTag);
        DataStream<String> startDs = flDs.getSideOutput(startTag);
        DataStream<String> actionDs = flDs.getSideOutput(actionTag);
        DataStream<String> displayDs = flDs.getSideOutput(displayTag);

        errDs.addSink(MyKafkaUtil.getFlinkKafkaProducer("dwd_traffic_error_log"));
        startDs.addSink(MyKafkaUtil.getFlinkKafkaProducer("dwd_traffic_start_log"));
        actionDs.addSink(MyKafkaUtil.getFlinkKafkaProducer("dwd_traffic_action_log"));
        displayDs.addSink(MyKafkaUtil.getFlinkKafkaProducer("dwd_traffic_display_log"));
        flDs.addSink(MyKafkaUtil.getFlinkKafkaProducer("dwd_traffic_page_log"));

        //7.触发执行
        env.execute();
    }
}
